package com.qupeng.demo.kafka.kafkaspringbootstreams.timewindow;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.CommonResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.controller.ProduceSuccessResponse;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.utils.KafkaService;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentCase;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentPresence;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentProfile;
import com.qupeng.demo.kafka.kafkaspringbootstreams.demo.CustomerComplaint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/message")
public class TimeWindowController {
    @Autowired
    private KafkaService<String, AgentCase> caseService;

    @PostMapping("/time-window-testing/data")
    public CommonResponse<ProduceSuccessResponse<String, String>> producePeter(@RequestBody Map<String, String> requestBody) {
        String agentIndex = requestBody.get("agentIndex");
        produce(requestBody.get("dataIndex"), agentIndex);

        return new CommonResponse<>("producer");
    }

    private void produce(String dataIndex, String agentIndex) {
        for (int i = 0; i < 10; i++) {
            AgentCase agentCaseHandling = new AgentCase("c" + dataIndex + "0" + i, "u00" + agentIndex, "HANDLING", "2024-04-27T18:0" + i + ":00+0800");
            this.<String, AgentCase>produce(caseService, agentCaseHandling, "time-window-testing-input", agentCaseHandling.getAgentId());
        }
    }

    private <K, V>CommonResponse<ProduceSuccessResponse<K, V>> produce(KafkaService<K, V> kafkaService, V messagePayload, String topic, K key) {
        try {
            SendResult<K, V> sendResult = kafkaService.sendToKafkaSync(topic, key, messagePayload);
            return new CommonResponse<>("producer", new ProduceSuccessResponse<K, V>(sendResult));
        } catch (ExecutionException | InterruptedException e) {
            return new CommonResponse<>("001-producer-controller", e.getMessage(), "producer");
        }
    }
}
